package com.niit.service

import com.niit.bean.Answer
import com.niit.utils.SparkUtil
import org.apache.spark.ml.recommendation.ALSModel
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.apache.spark.streaming.dstream.DStream

/*

  根据实时信息进行推荐， 将推荐的结果保存道数据库当中
 */
class EDURecommendService {

  val spark =SparkUtil.takeSpark()
  import spark.implicits._
  import org.apache.spark.sql.functions._

  def dataAnalysis(answer:DStream[Answer]): Unit ={
    answer.foreachRDD(rdd=>{
    //1.获取模型
    val model_path = "output/als_model/1716947383644"
    //2.加载模型
    val model: ALSModel = ALSModel.load(model_path)

    //由于寻模型的时候，不能有任何中文仔阳，所以在训练的时候我们将id进行了切割 学生ID_7 ==  7
    //将学生id转换为数字 这里UDF(用户自定义)方法。因为后续模型推荐中需要用到数字格式id
    val id2Int = udf((student_id:String)=>{
      //学生ID_7  对实时数据进行的学生id 都按照_进行切割 得到后面数字
      student_id.split("_")(1).toInt
    })

    //3.由于ALS模型传入的参数必须是DataFrame ,所以要将rdd转换成DataFrame
    val answerDF: DataFrame = rdd.toDF()

      //4.根据answerDF 得到学生id  只要数字
      //由于是针对学生进行推荐错题，所以先将student查询并切割出来
      val studentIdDF: DataFrame = answerDF.select(id2Int('student_id) as "student_id")

      //5.使用模型给学生推荐题目
      val recommendDF: DataFrame = model.recommendForUserSubset(studentIdDF, 10)
      //打印 recommendDF 数据结构 和内容
      recommendDF.printSchema()
      recommendDF.show(false)

      //6.处理推荐结果，取出用户id和 题目id.将题目拼成一个字符串
      val res: DataFrame = recommendDF.as[(Int, Array[(Int, Float)])].map(t => {
        val studentId: String = "学生ID_" + t._1
        val questionId = t._2.map("题目ID_" + _._1).mkString(",")
        (studentId, questionId)
      }).toDF("student_id", "recommendations")

      //7.将ansewerDF 和 处理后的推荐结果进行 合并
      val allInfoDF: DataFrame = answerDF.join(res, "student_id")
      //8.保存到数据库当中
      allInfoDF.write
        .format("jdbc")
        .option("url","jdbc:mysql://node1:3306/BD1?useUnicode=true&characterEncoding=utf8")
        .option("driver","com.mysql.jdbc.Driver")
        .option("user","root")
        .option("password","Niit@123")
        .option("dbtable","edu")
        .mode(SaveMode.Append)
        .save()
    })

  }
}
